package com.atguigu.flink.sql.window;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/**
 * Created by Smexy on 2023/4/11
 */
public class Demo2_DefineTimeSQL
{
    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();

        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        /*
               事件时间必须是TIMESTAMP 类型 或 TIMESTAMP_LTZ类型

                     2020-04-15 20:13:40.564 格式: TIMESTAMP(3) 类型
                     2020-04-15 20:13:40 格式: TIMESTAMP(0) 类型

                     1618989564564 格式:  TIMESTAMP_LTZ(3)
                     1618989564    格式:  TIMESTAMP_LTZ(0)
         */
        String createTableSql = " create table t1 ( id STRING, ts BIGINT , vc INT , " +
            "   pt  AS PROCTIME() , " +
            "   et  AS TO_TIMESTAMP_LTZ(ts,3)  ," +
            "   WATERMARK FOR et AS et - INTERVAL '0.001' SECOND " +
            "    )WITH (" +
            "  'connector' = 'filesystem'," +
            "  'path' = 'data/sensor.json'," +
            "  'format' = 'json'" +
            ") ";

        //建表
        tableEnvironment.executeSql(createTableSql);

        //查询
        tableEnvironment.sqlQuery(" select * from t1  ").execute().print();

    }

}
